package drds.data_migrate.task_group.task._runnable_;

import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.task_group.task.data_exchange.exchanger.Produce;
import drds.data_migrate.task_group.task.run.ReadTask;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Author(name = Authors.LI_YANG)
@Slf4j
public class ReadRunnable extends AbstractRunnable implements Runnable
{
    private Produce produce;
    @Setter
    @Getter
    private TaskConfig taskConfig;
    private ReadTask readTask;


    public void setProduce(Produce produce)
    {
        this.produce = produce;
    }

    public void run()
    {
        assert null != this.produce;

        //ReadTaskWrapper readTaskWrapper = null;
        readTask = new ReadTask(taskConfig);

        try
        {


            log.debug("wrapper reade starts to do init ...");


            this.readTask = new ReadTask(taskConfig);
            this.readTask.init();


            log.debug("wrapper reade starts to do prepare ...");


            log.debug("wrapper reade starts to read ...");


            this.readTask.startRead(produce);
            produce.terminate();


            log.debug("wrapper reade starts to do post ...");

            readTask.post();
            readTask.destroy();

            // automatic pushAllToRecordQueue
            // super.markSuccess(); 这里不能标记为成功，成功的标志由 writerRunner 来标志（否则可能导致 reade 先结束，而 write 还没有结束的严重 bug）
        } catch (Throwable e)
        {
            log.error("AbstractReader wrapper Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            log.debug("wrapper reade starts to do destroy ...");


        }
    }

    public void shutdown()
    {
        produce.shutdown();
    }
}
